/* 
 * File:   prepare_service.cpp
 * Author: james
 *
 * Created on 2011年4月2日, 下午2:23
 */

#ifndef PREPARE_SERVICE_CPP
#define	PREPARE_SERVICE_CPP

#include "prepare_service.h"
#include "message.h"
#include "singleton.h"
#include "msp_control.h"
#include <json/json.h>
#include <bits/stl_map.h>

namespace edm{
namespace prepare{

prepare_service_t::prepare_service_t(const char* bind, const std::string& host, const std::string& port, const std::string& msp_control):
        m_bind_host(bind), m_connection(host, port), m_msp_control_path(msp_control), m_msp_control_parser(':'), m_started(false),m_delivery_handler(0),
        m_io_work(new boost::asio::io_service::work(m_io_service)),m_timer(m_io_service), m_thread(0)
{
    logtrace((PREPARE, "prepare_service_t::prepare_service_t connect dispatcher:<%s:%s>", 
              host.c_str(), port.c_str()));
}

prepare_service_t::~prepare_service_t()
{
    logtrace((PREPARE, "prepare_service_t::prepare_service_t begin ..."));
    if(m_started) stop();
    logtrace((PREPARE, "prepare_service_t::prepare_service_t end ok."));
}

//! set init block msp
void prepare_service_t::block(std::vector<std::string>& blocks)
{
    m_blocks.assign(blocks.begin(), blocks.end());
}

//! start prepare service
int prepare_service_t::start(delivery_handler_t delivery_handler)
{
    logtrace((PREPARE, "prepare_service_t::start begin ..."));
    if(m_started)
    {
        logwarn((PREPARE, "prepare_service_t::start already started"));
        return 0;
    }

    if(m_msp_control_parser.open(m_msp_control_path.c_str()))
    {
        logerror((PREPARE, "prepare_service_t::start => open msp control <%s> failed",
                m_msp_control_path.c_str()));
        return -1;
    }
    
    //! start connection
    if(m_connection.start(boost::bind(&prepare_service_t::callback_handler, this, _1, _2),
                        boost::bind(&prepare_connection_t::callback_error, &m_connection, _1),
                        boost::bind(&prepare_service_t::callback_connect, this, _1)))
    {
        logerror((PREPARE, "prepare_service_t::start => connection start failed"));
        return -1;
    }
    
    m_delivery_handler = delivery_handler;
    //! start thread
    m_thread.reset(new boost::thread(
    boost::bind(&boost::asio::io_service::run, &m_io_service)));
    m_timer.expires_from_now(boost::posix_time::seconds(PREPARE_TIMER_TICK));
    m_timer.async_wait(boost::bind(&prepare_service_t::timer_i, this));
    
    //! set flag
    m_started = true;
    logtrace((PREPARE, "prepare_service_t::start end ok."));
    return 0;    
}

//! stop prepare service
int prepare_service_t::stop()
{
    logtrace((PREPARE, "prepare_service_t::stop begin ..."));
    if(!m_started)
    {
        logwarn((PREPARE, "prepare_service_t::stop already stopped"));
        return 0;
    }    
    
    std::map<std::string, msp_control_t*>::iterator it;
    for(it = m_msps_map.begin(); it != m_msps_map.end(); it++)
    {        
        m_msp_memory_pool.destroy(it->second);
    }
    
    m_connection.stop(true);
    
    m_timer.cancel();
    m_io_work.reset();
    m_thread->join();
    m_started = false;
    logtrace((PREPARE, "prepare_service_t::stop end ok."));
    return 0;
}

            
//! prepare connection connect callback handler
void prepare_service_t::callback_connect(edm::network::connection_ptr_t c)
{
    logtrace((PREPARE, "prepare_service_t::callback_connect begin ..."));
    m_connection.callback_connect(c);
    prepare_status_i(true);
}

//! prepare connection message callback handler
void prepare_service_t::callback_handler(edm::network::connection_ptr_t c, edm::network::message_t& m)
{
    logtrace((PREPARE, "prepare_service_t::callback_handler"));
    m_io_service.post(boost::bind(&prepare_service_t::callback_handler_i, this, c, m));
}

//! callback for delivery
void prepare_service_t::callback_delivery(delivery_t* delivery)
{
    logtrace((PREPARE, "prepare_service_t::callback_delivery"));
    m_io_service.post(boost::bind(&prepare_service_t::fsm_i, this, delivery));
}

//! msp block invoke
void prepare_service_t::msp_block(msp_control_t* msp, std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::msp_block"));
    m_io_service.post(boost::bind(&prepare_service_t::msp_block_i, this, msp, json));
}

//! msp unblock invoke
void prepare_service_t::msp_unblock(msp_control_t* msp, std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::msp_unblock"));
    m_io_service.post(boost::bind(&prepare_service_t::msp_unblock_i, this, msp, json));
}

//! prepare helo response
void prepare_service_t::prepare_status(bool with_prepare_id)
{
    logtrace((PREPARE, "prepare_service_t::prepare_status"));
    m_io_service.post(boost::bind(&prepare_service_t::prepare_status_i, this, with_prepare_id));    
}

//! msp status invoke
void prepare_service_t::msp_status(std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::msp_status"));
    m_io_service.post(boost::bind(&prepare_service_t::msp_status_i, this, json));    
}

//! prepare helo response
void prepare_service_t::prepare_missed_temp(edm_task_id_t taskid)
{
    logtrace((PREPARE, "prepare_service_t::prepare_status"));
    m_io_service.post(boost::bind(&prepare_service_t::prepare_missed_temp_i, this, taskid)); 
}

//! timer
void prepare_service_t::timer_i()
{
    logtrace((PREPARE, "prepare_service_t::timer_i"));
    worker_i();
    m_timer.expires_from_now(boost::posix_time::seconds(PREPARE_TIMER_TICK));
    m_timer.async_wait(boost::bind(&prepare_service_t::timer_i, this));
}

void prepare_service_t::worker_i()
{
    logtrace((PREPARE, "prepare_service_t::worker_i"));
    
    loginfo((PREPARE, "STATUS:<%s>: <work>:%d, <wait>:%d, <retry>:%d, <block>:%d, <stop>:%d",
            m_bind_host.c_str(),
            m_working_deliverys.size(),
            m_waitting_deliverys.size(),
            m_retrying_deliverys.size(),
            m_blocking_deliverys.size(),
            m_stopped_deliverys.size()));
    
    //! <stop> queue process
    if(!m_stopped_deliverys.empty())
    {
        delivery_t* delivery = m_stopped_deliverys.front();
        edm_task_id_t taskid = delivery->task_id;
        Json::Value json_status;
        Json::Value json_delivery;

        while(!m_stopped_deliverys.empty())
        {
            m_stopped_deliverys.pop_front();
            //! build json
            Json::Value send_result;
            send_result["addr"] = Json::Value(delivery->recipient);
            send_result["code"] = Json::Value((int)delivery->smtp_low_code);
            json_delivery.append(send_result);
            m_delivery_memory_pool.destroy(delivery);
            
            if(!m_stopped_deliverys.empty() && m_stopped_deliverys.front()->task_id != taskid) break;
            delivery = m_stopped_deliverys.front();
        }
        
        json_status["delivery"] = json_delivery;
        Json::FastWriter writer;
        std::string body = writer.write(json_status);  
        task_status_i(taskid, body);
    }
    
    time_t current = time(0);
    
    //! retry process
    delivery_list_t current_retrying = m_retrying_deliverys;m_retrying_deliverys.clear();
    while(!current_retrying.empty())
    {
        delivery_t* delivery = current_retrying.front();
        current_retrying.pop_front();
        if(delivery->next_retry_time > current)
           m_retrying_deliverys.push_back(delivery);
        else
           push_to_wait_i(delivery);
    }
    
    //! block process
    delivery_list_t current_blocking = m_blocking_deliverys;m_blocking_deliverys.clear();
    while(!current_blocking.empty())
    {
        delivery_t* delivery = current_blocking.front();
        current_blocking.pop_front();
        if(delivery->next_retry_time > current)
        {
            m_blocking_deliverys.push_back(delivery);
        }
        else
            push_to_work_i(delivery, true);//! blocking queue will not goto <wait> queue again
    }
    
    //! <wait> queue process
    delivery_list_t current_waiting = m_waitting_deliverys;m_waitting_deliverys.clear();
    while(!current_waiting.empty())
    {
        delivery_t* delivery = current_waiting.front();
        current_waiting.pop_front();
        
        //! task control
        if(pause(delivery))
        {
            m_waitting_deliverys.push_back(delivery);
            continue;
        }
        
        if(cancel(delivery))
        {
            delivery->state = edm::prepare::delivery_t::OPERATOR_DELETED;
            delivery->smtp_low_code = edm::common::OPERATOR_DELETED;
            push_to_stop_i(delivery);
            continue;
        }
        
        //!UNBLOCKED, EXCEED, BLOCKED
        switch(delivery->msp->status()) 
        {
            case msp_control_t::UNBLOCKED:
            {    
                push_to_work_i(delivery);
                break;
            }
            case msp_control_t::EXCEED:
            {    
                m_waitting_deliverys.push_back(delivery);
                break;
            }    
            case msp_control_t::BLOCKED:
            {    
                push_to_block_i(delivery);
                break;
            }
        }
    }
    
    //! clear pause, cancel
    m_pause_tasks.clear();
    m_cancel_tasks.clear();
    
    //! block msp clear
    std::set<msp_control_t*>::iterator it;
    for(it = m_block_msps.begin(); it != m_block_msps.end(); it++)
    {
        if(((*it)->last_block_time() + DELIVERY_BLOCK_INTERVAL) <= current)
        {
            (*it)->block(false);
            loginfo((PREPARE, "STATUS: <%s> unblock msp:<%s> for msp block timeout.",
                    m_bind_host.c_str(), (*it)->msp_name().c_str()));
        }
    }
    
    //! prepare spy info
    prepare_spy_i();
    
    loginfo((PREPARE, "STATUS:<%s>: <work>:%d, <wait>:%d, <retry>:%d, <block>:%d, <stop>:%d",
            m_bind_host.c_str(),
            m_working_deliverys.size(),
            m_waitting_deliverys.size(),
            m_retrying_deliverys.size(),
            m_blocking_deliverys.size(),
            m_stopped_deliverys.size()));
}

bool prepare_service_t::pause(delivery_t* d)
{
   return (m_pause_tasks.find(d->task_id) != m_pause_tasks.end()) ? true : false;
}

bool prepare_service_t::cancel(delivery_t* d)
{
   return (m_cancel_tasks.find(d->task_id) != m_cancel_tasks.end()) ? true : false;
}
            
//! delivery fsm
void prepare_service_t::fsm_i(delivery_t* d)
{
    logtrace((PREPARE, "prepare_service_t::fsm_i"));
    if(d)
    {
        this->m_working_deliverys.remove(d);
        
        switch((delivery_t::DSTATE)d->state)
        {
            case delivery_t::UNCOOK:
            {
                this->push_to_wait_i(d);break;
            }    
            case delivery_t::UNCOOK_TEMPERR:
            {
                this->push_to_wait_i(d);break;
            }
            case delivery_t::COOKED:
            {
                this->push_to_wait_i(d);break;
            }
            case delivery_t::COOKED_TEMPERR:
            {
                this->push_to_wait_i(d);break;
            }
            case delivery_t::DELIVERIED_TEMPERR:
            {
                this->push_to_retry_i(d);break;
            }
            case delivery_t::OPERATOR_DELETED:    
            {
                this->push_to_stop_i(d);break;
            }
            case delivery_t::DELIVERIED:
            {
                d->msp->decrement_concurrency(d->smtp_high_code, d->smtp_low_code);
                //if(d->smtp_high_code == edm::common::SMTP_RETRY)
                //    this->push_to_retry_i(d);
                //else
                this->push_to_stop_i(d);break;
            }
        }
    }
}

void prepare_service_t::push_to_work_i(delivery_t* d, bool force)
{
    logtrace((PREPARE, "prepare_service_t::push_to_work_i"));
    if(d)
    {
        loginfo((PREPARE, "WORK: delivery task:<%lld> from:<%s> => to:<%s> push <work> queue.",
                    d->task_id, d->from_addr.c_str(), d->recipient.c_str()));

        if(force)
        {
            m_working_deliverys.push_back(d);
            if(m_delivery_handler)
            {
                d->msp->increment_concurrency(1);
                m_delivery_handler(d);
            }
            else
                push_to_stop_i(d);
        }
        else
        {
            if(m_working_deliverys.size() < DELIVERY_CURRENCY_MAX)
            {
                m_working_deliverys.push_back(d);
                if(m_delivery_handler)
                {
                    d->msp->increment_concurrency(1);
                    m_delivery_handler(d);
                }
                else
                    push_to_stop_i(d);
            }
            else
            {
                loginfo((PREPARE, "WORK: delivery task:<%lld> from:<%s> => to:<%s> push to <wait> queue front, for exceed MAX currecy.",
                        d->task_id, d->from_addr.c_str(), d->recipient.c_str(), d->retry));
                this->m_waitting_deliverys.push_front(d);
            }
        }
        
        
    }
}

void prepare_service_t::push_to_wait_i(delivery_t* d)
{
    logtrace((PREPARE, "prepare_service_t::push_to_wait_i"));
    if(d)
    {
        loginfo((PREPARE, "WAIT: delivery task:<%lld> from:<%s> => to:<%s> push to <wait> queue, retrying count:<%d>.",
                    d->task_id, d->from_addr.c_str(), d->recipient.c_str(), d->retry));
        this->m_waitting_deliverys.push_back(d);
    }
}

void prepare_service_t::push_to_retry_i(delivery_t* d)
{
    logtrace((PREPARE, "prepare_service_t::push_to_retry_i"));
    if(d)
    {
        loginfo((PREPARE, "RETRY: delivery task:<%lld> from:<%s> => to:<%s> push to <retry> queue, retrying count:<%d>.",
                    d->task_id, d->from_addr.c_str(), d->recipient.c_str(), d->retry));
        if(++d->retry > DELIVERY_RETRY_MAX)
        {
            push_to_stop_i(d);
        }
        else
        {
            d->next_retry_time = time(0) + DELIVERY_RETRY_INTERVAL;
            this->m_retrying_deliverys.push_back(d);
        }
    }
}

void prepare_service_t::push_to_block_i(delivery_t* d)
{
    logtrace((PREPARE, "prepare_service_t::push_to_block_i"));
    if(d)
    {
        loginfo((PREPARE, "BLOCK: delivery task:<%lld> from:<%s> => to:<%s> push to <block> queue.",
                    d->task_id, d->from_addr.c_str(), d->recipient.c_str()));
        d->next_retry_time = d->msp->last_block_time() + DELIVERY_BLOCK_INTERVAL;
        this->m_blocking_deliverys.push_back(d);
    }
}

void prepare_service_t::push_to_stop_i(delivery_t* d)
{
    logtrace((PREPARE, "prepare_service_t::push_to_stop_i"));
    if(d)
    {
        loginfo((PREPARE, "STOP: delivery task:<%lld> from:<%s> => to:<%s> push to <stop> queue, smtp code:<%d>.",
                    d->task_id, d->from_addr.c_str(), d->recipient.c_str(), d->smtp_high_code + d->smtp_low_code));
        m_working_deliverys.remove(d);
        m_stopped_deliverys.push_back(d);            
    }
}

//! prepare helo response
void prepare_service_t::prepare_status_i(bool with_prepare_id)
{
    logtrace((PREPARE, "prepare_service_t::prepare_status_i begin ..."));
    /*CMD_C2S_STATUS*/
    if(with_prepare_id)
    {
        Json::Value json_helo;
        Json::Value json_block;
        json_helo["prepare_id"] = Json::Value(m_bind_host);
        std::vector<std::string>::iterator it;
        for(it = m_blocks.begin(); it != this->m_blocks.end(); it++)
        {
            json_block.append(Json::Value(*it));
        }
        json_helo["block"] = json_block;
        Json::FastWriter writer;
        std::string body = writer.write(json_helo);  
        m_connection.async_write(edm::common::CMD_C2S_STATUS, 0, body);
    }
    logtrace((PREPARE, "prepare_service_t::prepare_status_i end ok."));
}

//! prepare spy response
void prepare_service_t::prepare_spy_i()
{
    logtrace((PREPARE, "prepare_service_t::prepare_status_i begin ..."));
        
    Json::Value json_spy;
    json_spy["host"] = Json::Value(m_bind_host);
    json_spy["work"] = Json::Value((int)this->m_working_deliverys.size());
    json_spy["wait"] = Json::Value((int)this->m_waitting_deliverys.size());
    json_spy["retry"] = Json::Value((int)this->m_retrying_deliverys.size());
    json_spy["block"] = Json::Value((int)this->m_blocking_deliverys.size());
    json_spy["stopped"] = Json::Value((int)this->m_stopped_deliverys.size());
    Json::FastWriter writer;
    std::string body = writer.write(json_spy);  
    m_connection.async_write(edm::common::CMD_C2S_SPY, 0, body);
    logtrace((PREPARE, "prepare_service_t::prepare_status_i end ok."));
}
            
//! prepare helo response
void prepare_service_t::prepare_missed_temp_i(edm_task_id_t taskid)
{
    logtrace((PREPARE, "prepare_service_t::prepare_missed_temp_i begin ..."));
    /*CMD_C2S_TEMPLATE_MISSED*/
    m_connection.async_write(edm::common::CMD_C2S_TEMPLATE_MISSED, taskid, "");
    logtrace((PREPARE, "prepare_service_t::prepare_missed_temp_i end ok."));   
}

//! prepare connection message callback handler
void prepare_service_t::callback_handler_i(edm::network::connection_ptr_t c, edm::network::message_t& m)
{
    logtrace((PREPARE, "prepare_service_t::callback_handler_i begin ..."));
    /*
            CMD_S2C_STATUS = 1,
            CMD_C2S_STATUS,
            CMD_C2S_BLOCK,
            CMD_C2S_UNBLOCK,
            CMD_S2C_TASK_DELIVERY = 11,
            CMD_C2S_MSP_RESPONSE,
            CMD_C2S_TEMPLATE_MISSED,
            CMD_S2C_TASK_PAUSE,
            CMD_S2C_TASK_START,
            CMD_S2C_TASK_DELETE,
     */
    switch(m.header.command)
    {
        case edm::common::CMD_S2C_STATUS:
        {
            //! json format: null
            prepare_status(false);
            break;
        }
        case edm::common::CMD_S2C_TASK_DELIVERY:
        {
            //! json format:  {"from":{"name":"", "addr":""}, "to":"", "replace":["",""]}
            //! TODO::
            Json::Reader reader;
            Json::Value  value;
            if (!reader.parse(m.body, value)) 
            {
                logwarn((PREPARE, "prepare_service_t::callback_handler_i => json parse failed."));
                return;
            }
            
            if(value["from"].isNull() || 
               value["to"].isNull())
            {
                logwarn((PREPARE, "prepare_service_t::callback_handler_i => json absent some keys."));
                return;
            }
            
            if (value["from"]["name"].isNull() || value["from"]["addr"].isNull())
            {
                logwarn((PREPARE, "prepare_service_t::callback_handler_i => json absent <from> sub keys."));
                return;
            }

            std::string domain;
            std::string to = value["to"].asString();
            std::string::size_type pos = to.find('@');
            if ((pos == std::string::npos) || (pos == to.size()-1)) {
                logwarn((PREPARE, "prepare_service_t::callback_handler_i => json <to:%s> field not right.", to.c_str()));
                return;
            }
            domain = to.substr(pos+1);
            
            //! msp start delivery
            std::string from_name = value["from"]["name"].asString();
            std::string from_addr = value["from"]["addr"].asString();
            
            std::vector<std::string> replace;
            if(!value["replace"].isNull())
            {
                for(size_t i = 0; i < value["replace"].size(); i++)
                {
                    replace.push_back(value["replace"][i].asString());
                }
            }
            logdebug((PREPARE, "from:<name:%s, addr:%s> => to:<%s>", 
                      from_name.c_str(), from_addr.c_str(), to.c_str()));
            
            //! alloc delivery
            delivery_t* delivery = m_delivery_memory_pool.construct((edm::prepare::edm_task_id_t)m.header.reserve, (edm::prepare::msp_control_t*)0);
            if(!delivery)
            {
                logwarn((PREPARE, "PREPARE: construct delivery failed, no more memory!!!"));
                //! no resource anymore, just back send to the dispatcher
                c->async_write((edm::common::COMMANDER)m.header.command, m.header.reserve, m.body);
                return;
            }
               
            delivery->from_addr = from_addr;
            delivery->from_name = from_name;
            delivery->recipient = to;
            delivery->replace = replace;
            if(!value["top"].isNull())
            {
               delivery->content_top = value["top"].asString(); 
            }
            if(!value["bottom"].isNull())
            {
               delivery->content_bottom = value["bottom"].asString(); 
            }
            
            //! find msp
            std::map<std::string, msp_control_t*>::iterator it;
            msp_control_t* msp = 0;
            it = m_msps_map.find(domain);
            if(it == m_msps_map.end())
            {
                int concurrency;
                m_msp_control_parser.get_int_value(domain.c_str(), concurrency, MSP_DEFAULT_CURRENCY);
                msp = m_msp_memory_pool.construct(*this, domain, concurrency);
                if(!msp)
                {
                    logwarn((PREPARE, "PREPARE: construct msp control failed, no more memory!!!"));
                    //! no resource anymore, just back send to the dispatcher
                    c->async_write((edm::common::COMMANDER)m.header.command, m.header.reserve, m.body);
                    m_delivery_memory_pool.destroy(delivery);
                    return; 
                }
                
                m_msps_map[domain] = msp;
                loginfo((PREPARE, "MSP: create new msp :<%s> concurrency:<%d>, msp current counting:<%d>",
                          domain.c_str(), concurrency, m_msps_map.size() + 1));
            }
            else
                msp = it->second;
            delivery->msp = msp;
            
            fsm_i(delivery);
            worker_i();
            break;
        }
        case edm::common::CMD_S2C_TASK_PAUSE:
        {
            m_pause_tasks.insert(m.header.reserve);
            break;
        }
        case edm::common::CMD_S2C_TASK_START:
        {
            m_pause_tasks.erase(m.header.reserve);
            m_cancel_tasks.erase(m.header.reserve);
            break;
        }
        case edm::common::CMD_S2C_TASK_DELETE:
        {
            m_cancel_tasks.insert(m.header.reserve);
            break;
        }
        default:
        {
            logwarn((PREPARE, "prepare_service_t::callback_handler_i not right command:<%d>.",
                    m.header.command));
            return;
        }
    }
    logtrace((PREPARE, "prepare_service_t::callback_handler_i end ok."));
}

//! msp status invoke
void prepare_service_t::msp_status_i(std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::msp_status_i begin ..."));
    /*CMD_C2S_STATUS*/
    m_connection.async_write(edm::common::CMD_C2S_STATUS, 0, json);
    logtrace((PREPARE, "prepare_service_t::msp_status_i end ok."));    
}
            
//! msp block invoke
void prepare_service_t::msp_block_i(msp_control_t* msp, std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::msp_block_i begin ..."));
    m_block_msps.insert(msp);
    /*CMD_C2S_BLOCK*/
    m_connection.async_write(edm::common::CMD_C2S_BLOCK, 0, json);
    logtrace((PREPARE, "prepare_service_t::msp_block_i end ok."));
}

//! msp unblock invoke
void prepare_service_t::msp_unblock_i(msp_control_t* msp, std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::msp_unblock_i begin ..."));
    m_block_msps.erase(msp);
    /*CMD_C2S_UNBLOCK*/
    m_connection.async_write(edm::common::CMD_C2S_UNBLOCK, 0, json);
    logtrace((PREPARE, "prepare_service_t::msp_unblock_i end ok."));
}

//! msp_timer invoke
void prepare_service_t::task_status_i(edm_task_id_t taskid, std::string& json)
{
    logtrace((PREPARE, "prepare_service_t::task_status_i begin ..."));
    /*CMD_C2S_MSP_RESPONSE*/
    m_connection.async_write(edm::common::CMD_C2S_MSP_RESPONSE, taskid, json);
    logtrace((PREPARE, "prepare_service_t::task_status_i end ok."));
}
        
}}

#endif	/* PREPARE_SERVICE_CPP */

